package com.zyb.dfs.datanode.server;


import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import com.zyb.dfs.namenode.rpc.model.*;
import com.zyb.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;

import static com.zyb.dfs.datanode.server.DataNodeConfig.*;

/**
 * 负责跟一组NameNode中的某一个进行通信的线程组件
 *
 * @author dhh
 */
public class NameNodeRpcClient {

    private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode;

    public NameNodeRpcClient() {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();
        this.namenode = NameNodeServiceGrpc.newBlockingStub(channel);
        System.out.println("跟NameNode的" + NAMENODE_PORT + "端口建立连接......");
    }

    /**
     * 启动OfferService组件
     */
    public void start() throws Exception {
        // 直接使用两个ServiceActor组件分别向主备两个NameNode节点进行注册
        register();
        // 开始发送心跳
        // startHeartbeat();
    }

    /**
     * 向自己负责通信的那个NameNode进行注册
     */
    public Boolean register() throws Exception {
//		Thread registerThread = new RegisterThread();
//		registerThread.start();
//		registerThread.join();
        RegisterRequest request = RegisterRequest.newBuilder()
                .setIp(DATANODE_IP)
                .setHostname(DATANODE_HOSTNAME)
                .setNioPort(NIO_PORT)
                .build();
        RegisterResponse response = namenode.register(request);
        System.out.println("完成向NameNode的注册，响应消息为：" + response.getStatus());
        if (response.getStatus() == 1) {
            return true;
        }
        return false;
    }

    /**
     * 发送心跳
     * @throws Exception
     */
    public HeartbeatResponse heartbeat() throws Exception {
        HeartbeatRequest request = HeartbeatRequest.newBuilder()
                .setIp(DATANODE_IP)
                .setHostname(DATANODE_HOSTNAME)
                .setNioPort(NIO_PORT)
                .build();
        return namenode.heartbeat(request);
    }

    /**
     * 通知Master节点自己收到了一个文件的副本
     *
     * @param filename
     * @throws Exception
     */
    public void informReplicaReceived(String filename) throws Exception {
        InformReplicaReceivedRequest request = InformReplicaReceivedRequest.newBuilder()
                .setFilename(filename)
                .setHostname(DATANODE_HOSTNAME)
                .setIp(DATANODE_IP)
                .build();
        namenode.informReplicaReceived(request);
    }

    /**
     * 上报全量的存储信息
     */
    public void reportCompleteStorageInfo(StorageInfo storageInfo) {
        ReportCompleteStorageInfoRequest request = ReportCompleteStorageInfoRequest.newBuilder()
                .setIp(DATANODE_IP)
                .setHostname(DATANODE_HOSTNAME)
                .setFilenames(new JSONArray(storageInfo.getFilenames()).toString())
                .setStoredDataSize(storageInfo.getStoredDataSize())
                .build();

        namenode.reportCompleteStorageInfo(request);
    }

    /**
     * 负责注册的线程
     * @author zhonghuashishan
     *
     */
//	class RegisterThread extends Thread {
//
//		@Override
//		public void run() {
//			try {
//				// 发送rpc接口调用请求到NameNode去进行注册
//				// 在这里进行注册的时候会提供哪些信息过去呢？
//				// 比如说当前这台机器的ip地址、hostname，这两个东西假设是写在配置文件里的
//				// 我们写代码的时候，主要是在本地来运行和测试，有一些ip和hostname，就直接在代码里写死了
//				// 大家后面自己可以留空做一些完善，你可以加一些配置文件读取的代码
//				// 通过RPC接口发送到NameNode他的注册接口上去
//
//				RegisterRequest request = RegisterRequest.newBuilder()
//						.setIp(DATANODE_IP)
//						.setHostname(DATANODE_HOSTNAME)
//						.setNioPort(NIO_PORT)
//						.build();
//				namenode.register(request);
//
//				System.out.println("完成向NameNode的注册......");
//			} catch (Exception e) {
//				e.printStackTrace();
//			}
//		}
//
//	}

    /**
     * 负责心跳的线程
     *
     * @author zhonghuashishan
     */
//    class HeartbeatThread extends Thread {
//
//        @Override
//        public void run() {
//            try {
//                System.out.println("定时心跳线程启动......");
//
//                while (true) {
//                    // 通过RPC接口发送到NameNode他的注册接口上去
//                    HeartbeatRequest request = HeartbeatRequest.newBuilder()
//                            .setIp(DATANODE_IP)
//                            .setHostname(DATANODE_HOSTNAME)
//                            .setNioPort(NIO_PORT)
//                            .build();
//                    namenode.heartbeat(request);
//
//                    Thread.sleep(30 * 1000); // 每隔30秒发送一次心跳到NameNode上去
//                }
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        }
//
//    }

}
